/*
 * BSM6.0
 *
 * Copyright 2012-2013 (C) SINODATA CO., LTD. All Rights Reserved.
 */
package com.sinodata.bsm.center.engine.property;

import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;

import com.sinodata.bsm.center.cache.CollectTaskCache;
import com.sinodata.bsm.center.cache.PropertyValueCache;
import com.sinodata.bsm.center.engine.event.EventEngine;
import com.sinodata.bsm.center.util.ThreadPool;
import com.sinodata.bsm.center.util.spring.SpringContextHolder;
import com.sinodata.bsm.common.constants.DataTypeConstants;
import com.sinodata.bsm.common.vo.CollectTask;
import com.sinodata.bsm.common.vo.ResPropertyValue;

/**
 * <p>
 * Description: 指标数据处理分发中心，分多线程处理，但一个资源的数据不能同时处理
 * </p>
 *
 * @author tangli
 * @version 1.0

 * <p>
 * History: 
 *
 * Date                     Author         Version     Description
 * ---------------------------------------------------------------------------------
 * 2012-6-13 下午1:58:12          tangli         1.0         To create
 * </p>
 *
 * @since 
 * @see     
 */
public class PropertyValueDistributeCenter {
    private static PropertyValueDistributeCenter instance = new PropertyValueDistributeCenter();
    private final LinkedBlockingQueue<ResPropertyValue> queue = new LinkedBlockingQueue<ResPropertyValue>();
    private final Set<Long> register = new HashSet<Long>();
    private final Object lock = new Object();

    private final PropertyValueCache propertyValueCache = SpringContextHolder.getBean(PropertyValueCache.class);
    private final CollectTaskCache collectTaskCache = SpringContextHolder.getBean(CollectTaskCache.class);

    /**
     * 
     */
    private PropertyValueDistributeCenter() {
        Collections.synchronizedSet(register);
        new QueueListener().start();
    }

    public static PropertyValueDistributeCenter getInstance() {
        return instance;
    }

    protected synchronized void checkIn(Long resId) {
        register.add(resId);
    }

    protected synchronized void checkOut(Long resId) {
        synchronized (lock) {
            register.remove(resId);
            lock.notifyAll();
        }
    }

    public void dispose(ResPropertyValue value) {

        Long resId = value.getResId();
        if (register.contains(resId)) {
            queue.add(value);
        } else {
            checkIn(resId);
            ThreadPool.getInstance().execute(new PropertyValueWorker(value));
        }
    }

    class QueueListener extends Thread {
        public QueueListener() {
        }

        @Override
        public void run() {
            while (true) {
                ResPropertyValue value = null;
                try {
                    value = queue.take();
                } catch (InterruptedException e) {
                }
                if (value != null) {
                    Long resId = value.getResId();
                    while (register.contains(resId)) {
                        synchronized (lock) {
                            try {
                                lock.wait();
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }
                    }
                    checkIn(resId);
                    ThreadPool.getInstance().execute(new PropertyValueWorker(value));
                }
            }
        }
    }

    class PropertyValueWorker implements Runnable {
        private final ResPropertyValue value;

        public PropertyValueWorker(ResPropertyValue value) {
            this.value = value;
        }

        @Override
        public void run() {
            Long resId = value.getResId();

            try {
                cache(value);
                persistent(value);
                //交给事件引擎处理
                Long[] propIds = new Long[value.getValues().length];
                for (int i = 0; i < value.getValues().length; i++) {
                    String v = (value.getValues()[i]).value;
                    if (v == null || v.equals(DataTypeConstants.DATA_NOT_AVAILABLE)) {
                        propIds[i] = null;//置空，事件引擎不会处理
                    } else {
                        propIds[i] = value.getValues()[i].propId;
                    }
                }
                EventEngine.matches(resId, propIds);
                //end 事件引擎处理
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                checkOut(resId);
            }
        }

        private void cache(ResPropertyValue value) {
            Long resId = value.getResId();
            ResPropertyValue.PropValue[] values = value.getValues();
            for (int i = 0; i < values.length; i++) {
                propertyValueCache.putValue(resId, values[i].propId, values[i].value, value.getTime());
            }
        }

        private void persistent(ResPropertyValue value) {
            if (value.isTemporary()) {
                return;
            }
            Long taskId = value.getTaskId();
            if (taskId != null) {
                if (-1 == taskId.longValue()) { //tivoli集成的指标 直接入库保存
                    PropertyValuePersistentThread.getInstance().add(value);
                }
                if (-2 == taskId.longValue()) {//网络集成的指标
                    PropertyValuePersistentThread.getInstance().add(value);
                } else {
                    CollectTask collectTask = collectTaskCache.get(taskId);
                    if (collectTask != null && collectTask.getIsSave() == 1) {
                        PropertyValuePersistentThread.getInstance().add(value);
                    }
                }
            }
        }
    }
}
